<?php
/**
 * MQ操作类
 * 延期队列，参考：https://www.cnblogs.com/-mrl/p/11114116.html
 * rabbitmq-plugins enable rabbitmq_delayed_message_exchange 启用延迟插件
 * type = x-delayed-message
 * setArguments(['x-delayed-type' => 'direct'])
 * $exchange->publish($message, $routeKey, AMQP_NOPARAM, ['headers' => ['x-delay'=> 1000]]);
 */
declare(strict_types=1);

namespace Libraries\Connections\Drivers;

use AMQPChannelException;
use AMQPConnection;
use AMQPChannel;
use AMQPConnectionException;
use AMQPEnvelopeException;
use AMQPExchange;
use AMQPExchangeException;
use AMQPQueue;
use AMQPQueueException;
use Exception;
use Libraries\Config;

class Amqp
{
    private $objConn = null; // 连接
    private $objChannel = null; // 通道
    
    /**
     * 构造函数
     * @param array $arrConfig
     * @throws Exception
     */
    public function __construct(array $arrConfig = [])
    {
        if (!extension_loaded('amqp')) { // 检查扩展是否已加载
            throw new Exception('extension "amqp" not exists');
        }
        $arrConfig || $arrConfig = Config::get('rabbitmq');
        $arrConfig && $this->objConn = new AMQPConnection($arrConfig);
        if (!$this->objConn || !$this->objConn->connect()) {
            throw new Exception('can not connect to amqp server');
        }
        $this->objChannel = new AMQPChannel($this->objConn);
        if (!$this->objChannel->isConnected()) {
            throw new Exception('can not connect to amqp channel');
        }
    }

    /**
     * 设置QOS
     *
     * @param int $intSize
     * @param int $intCount
     * @throws AMQPConnectionException
     */
    public function setChannelQos(int $intSize, int $intCount): void
    {
        if ($this->objChannel && $this->objChannel->isConnected()) {
            $this->objChannel->qos($intSize, $intCount);
        }
    }
    
    /**
     * 析构函数
     */
    public function __destruct()
    {
        $this->close();
    }

    /**
     * 重新连接，如果返回True，说明无需重连，或已重连成功
     *
     * @return bool
     * @throws AMQPConnectionException
     */
    public function reConnect(): bool
    {
        if (!$this->objConn || !$this->objChannel) {
            return false;
        }
        if ($this->objConn->isConnected() && $this->objChannel->isConnected()) {
            return true;
        }
        if ($this->objConn->reconnect()) {
            $this->objChannel = new AMQPChannel($this->objConn);
            return $this->objChannel->isConnected();
        }
        return false;
    }
    
    /**
     * 关闭连接
     * @return bool
     */
    public function close(): bool
    {
        $this->objConn && $this->objConn->disconnect();
        return true;
    }

    /**
     * 获取交换机
     *
     * @param string $strExchangeName
     * @param string $strType AMQP_EX_TYPE_DIRECT（直接投递）、AMQP_EX_TYPE_FANOUT、AMQP_EX_TYPE_HEADERS、AMQP_EX_TYPE_TOPIC
     * @param array $arrArguments
     * @param int $intFlag AMQP_NOPARAM、AMQP_DURABLE（持久化）、AMQP_PASSIVE、AMQP_AUTODELETE，可以多项并用
     * @return AMQPExchange|null
     * @throws AMQPChannelException
     * @throws AMQPConnectionException
     * @throws AMQPExchangeException
     */
    public function getExchange(
        string $strExchangeName,
        string $strType = AMQP_EX_TYPE_DIRECT,
        array $arrArguments = [],
        int $intFlag = AMQP_DURABLE
    ): ?AMQPExchange {
        if (!$this->objChannel) {
            return null;
        }
        $objExchange = new AMQPExchange($this->objChannel);
        $objExchange->setName($strExchangeName);
        $objExchange->setType($strType);
        $arrArguments && $objExchange->setArguments($arrArguments);
        $objExchange->setFlags($intFlag);
        $objExchange->declareExchange();
        return $objExchange;
    }

    /**
     * 设置队列
     *
     * @param string $strQueueName
     * @param array $arrArguments
     * @param int $intFlag AMQP_DURABLE（持久化）、AMQP_PASSIVE、AMQP_EXCLUSIVE、AMQP_AUTODELETE，可以多项并用
     * @return AMQPQueue|null
     * @throws AMQPChannelException
     * @throws AMQPConnectionException
     * @throws AMQPQueueException
     */
    public function getQueue(string $strQueueName, array $arrArguments = [], int $intFlag = AMQP_DURABLE): ?AMQPQueue
    {
        if (!$this->objChannel) {
            return null;
        }
        $objQueue = new AMQPQueue($this->objChannel);
        $objQueue->setName($strQueueName);
        $arrArguments && $objQueue->setArguments($arrArguments);
        $objQueue->setFlags($intFlag);
        $objQueue->declareQueue();
        return $objQueue;
    }

    /**
     * 取队列中未处理消息的个数
     *
     * @param AMQPQueue $objQueue
     * @return int
     * @throws AMQPChannelException
     * @throws AMQPConnectionException
     */
    public function getQueueSize(AMQPQueue $objQueue): int
    {
        return $objQueue->declareQueue();
    }

    /**
     * 删除一个队列及其内容
     *
     * @param AMQPQueue $objQueue
     * @return int
     * @throws AMQPChannelException
     * @throws AMQPConnectionException
     */
    public function delQueue(AMQPQueue $objQueue): int
    {
        return $objQueue->delete();
    }

    /**
     * 清空队列中的内容
     *
     * @param AMQPQueue $objQueue
     * @return bool
     * @throws AMQPChannelException
     * @throws AMQPConnectionException
     */
    public function clearQueue(AMQPQueue $objQueue): bool
    {
        return $objQueue->purge();
    }

    /**
     * 按路由绑定队列到交换机
     *
     * @param AMQPQueue $objQueue
     * @param string $strExchangeName
     * @param string $strRouteKey
     * @return bool
     * @throws AMQPChannelException
     * @throws AMQPConnectionException
     */
    public function bindRoute(AMQPQueue $objQueue, string $strExchangeName, string $strRouteKey): bool
    {
        if (strlen($strRouteKey) && strlen($strExchangeName)) {
            return $objQueue->bind($strExchangeName, $strRouteKey);
        }
        return false;
    }

    /**
     * 解绑路由
     *
     * @param AMQPQueue $objQueue
     * @param string $strExchangeName
     * @param string $strRouteKey
     * @return bool
     * @throws AMQPChannelException
     * @throws AMQPConnectionException
     */
    public function unbindRoute(AMQPQueue $objQueue, string $strExchangeName, string $strRouteKey): bool
    {
        if (strlen($strRouteKey) && strlen($strExchangeName)) {
            return $objQueue->unbind($strExchangeName, $strRouteKey);
        }
        return false;
    }

    /**
     * 通过交换机发送消息
     *
     * @param AMQPExchange $objExchange
     * @param string $strMsg
     * @param string $strRouteKey
     * @param array $arrAttr
     * @param int $intFlag
     * @return bool
     * @throws AMQPChannelException
     * @throws AMQPConnectionException
     * @throws AMQPExchangeException
     */
    public function putMsg(
        AMQPExchange $objExchange,
        string $strMsg,
        string $strRouteKey,
        array $arrAttr = [],
        int $intFlag = AMQP_NOPARAM
    ): bool {
        if (empty($strRouteKey)) {
            return false;
        }
        return $objExchange->publish($strMsg, $strRouteKey, $intFlag, $arrAttr);
    }

    /**
     * 获取一条消息
     *
     * @param AMQPQueue $objQueue
     * @return array|null
     * @throws AMQPChannelException
     * @throws AMQPConnectionException
     */
    public function getMsg(AMQPQueue $objQueue): ?array
    {
        $objAMQPEnvelope = $objQueue->get(); // 不要传参AMQP_AUTOACK
        if (!$objAMQPEnvelope) {
            return null;
        }
        return [
            'id' => $objAMQPEnvelope->getDeliveryTag(),
            'msg' => $objAMQPEnvelope->getBody(),
        ];
    }

    /**
     * 确认消息已处理
     *
     * @param AMQPQueue $objQueue
     * @param int $intMsgId
     * @param int $intFlag
     * @return bool
     * @throws AMQPChannelException
     * @throws AMQPConnectionException
     */
    public function ackMsg(AMQPQueue $objQueue, int $intMsgId, int $intFlag = AMQP_NOPARAM): bool
    {
        return $objQueue->ack($intMsgId, $intFlag);
    }

    /**
     * 回调方式消费队列（慎用）
     *
     * @param AMQPQueue $objQueue
     * @param callable $callable
     * @param int $intFlag
     * @throws AMQPChannelException
     * @throws AMQPConnectionException
     * @throws AMQPEnvelopeException
     */
    public function consumeMsg(AMQPQueue $objQueue, callable $callable, int $intFlag = AMQP_NOPARAM): void
    {
        $objQueue->consume($callable, $intFlag);
    }
}
